package site.yunnong.atvris.recommend.offline.spark.featureing

import org.apache.log4j.{Level, Logger}
import org.apache.spark.ml.feature.OneHotEncoderEstimator
import org.apache.spark.{SparkConf, sql}
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.functions.col

/**
 *
 *
 * @author zjh
 * @date 2021/9/15 16:45
 */
object FeatureEngineering {
  
  def oneHotEncoder(samples:DataFrame, inputCol: String, middleCol: String, outputCol: String): Unit ={
    // samples 样本集中的每条数据代表一部视频的信息，其中VideoId为视频id
    val samplesWithIdNumber = samples.withColumn(middleCol, col(inputCol).cast(sql.types.IntegerType))
    
    // 利用Spark的机器学习库Spark MLlib创建One-hot编码器
    val oneHotEncoder = new OneHotEncoderEstimator()
      .setInputCols(Array(middleCol))
      .setOutputCols(Array(outputCol))
      .setDropLast(false);
    
    // 训练one-hot编码器，并完成从id特征到one-hot向量的转换
    val oneHotEncoderSamples = oneHotEncoder.fit(samplesWithIdNumber).transform(samplesWithIdNumber);
    // 打印最终样本的数据结构
    oneHotEncoderSamples.printSchema();
    
    oneHotEncoderSamples.show(10);
  }
  
  def multiHotEncoder(samples: DataFrame, inputCol:String, middleCol:String, outputCol: String): Unit = {
    
  }

  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.ERROR)
    val conf = new SparkConf()
      .setMaster("local")
      .setAppName("featureEngineering")
      .set("spark.submit.deployMode", "client");
    
    val spark = SparkSession.builder().config(conf).getOrCreate();
    val movieResourcesPath = this.getClass.getResource("/sampledata/movies.csv");
    val movieSamples = spark.read.format("csv").option("header", "true").load(movieResourcesPath.getPath);
    println("Raw Movie Samples:")
    movieSamples.printSchema()
    movieSamples.show(10)
    println("OneHotEncoder:Example:")
    oneHotEncoder(movieSamples, "movieId", "movieIdNumber", "movieIdVector");
  }
}
